package com.hup.minicontroller.common.thread.component;

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 效率型任务执行器;
 * 简化版的Tomcat.ThreadPoolExecutor,只移植了[先涨线程池再排队]的核心,没有了Tomcat的其他逻辑
 * <pre>
 * 在任务较多的情况下增长线程池方式:
 *  JDK是先排队再涨线程池:更注重 减少线程创建等过程的资源消耗问题;非常注重线程复用,减少线程创建
 *  Tomcat则是先涨线程池再排队:更注重 并发任务的执行效率;只要有需要就创建线程,尽量复用线程
 *
 * Tomcat.ThreadPoolExecutor
 *  .原文注释 Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient method...
 * JDK.ThreadPoolExecutor
 *  .当有任务提交的时候，会创建核心线程去执行任务（即使有核心线程空闲）；
 *  .当核心线程数达到corePoolSize时，后续提交的都会进BlockingQueue中排队；
 *  .当BlockingQueue满了(offer失败)，就会创建临时线程(临时线程空闲超过一定时间后，会被销毁)；
 *  .当线程总数达到maximum时，后续提交的任务都会被RejectedExecutionHandler拒绝。
 *
 * 参考资料:
 * 简化版的Tomcat执行器
 *  https://www.jianshu.com/p/98f530bd93b0
 *  org.apache.tomcat.util.threads.ThreadPoolExecutor
 *  org.apache.tomcat.util.threads.TaskQueue
 * JDK和Tomcat中的线程池
 *  https://blog.csdn.net/Holmofy/article/details/81610481
 * </pre>
 *
 * @author hugan
 * @apiNote 只能照搬
 * @date 2020/5/29
 */
public class EfficientExecutor extends ThreadPoolExecutor {

    /**
     * 已提交的任务数
     */
    private final AtomicInteger submittedCount = new AtomicInteger(0);
    /**
     * 等待队列容量
     */
    private int queueSize;
    /**
     * 允许提交的最大任务数=最大线程数+等待队列容量
     */
    private int maxSubmittedCount;

    /**
     * 非核心线程[1分钟回收]的执行器
     */
    public EfficientExecutor(int corePoolSize, int maximumPoolSize, int queueSize, String threadNamePrefix,
            String executorDesc) {
        this(corePoolSize, maximumPoolSize, queueSize,
                1, TimeUnit.MINUTES,//非核心线程[1分钟回收]
                threadNamePrefix, executorDesc);
    }

    public EfficientExecutor(int corePoolSize, int maximumPoolSize, int queueSize,
            long keepAliveTime, TimeUnit unit,
            String threadNamePrefix, String executorDesc) {
        super(corePoolSize, maximumPoolSize,
                keepAliveTime, unit,
                new EfficientTaskQueue(),
                new PrefixAbleThreadFactory(threadNamePrefix),//实现修改线程名前缀
                new SimpleRejectedHandler(executorDesc));//达到最大任务时抛出异常
        ((EfficientTaskQueue) getQueue()).setParent(this);

        this.queueSize = queueSize;
        maxSubmittedCount = maximumPoolSize + queueSize;
    }

    /**
     * 实现允许构造对象后二次修改
     */
    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
        maxSubmittedCount = getMaximumPoolSize() + queueSize;
    }

    /**
     * 实现允许构造对象后二次修改
     */
    @Override
    public void setMaximumPoolSize(int maximumPoolSize) {
        super.setMaximumPoolSize(maximumPoolSize);
        maxSubmittedCount = maximumPoolSize + queueSize;
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedCount.decrementAndGet();
        //不了解Tomcat此处的后续逻辑
    }

    public int getSubmittedCount() {
        return submittedCount.get();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(Runnable command) {
        execute(command, 0, TimeUnit.MILLISECONDS);
    }

    public void execute(Runnable command, long timeout, TimeUnit unit) {
        /*
        本类和EfficientTaskQueue的代码,搬自[简书文章]和[Tomcat源码],
        已尽量从[Tomcat]中移植,因为[简书]的好像有点...
        但智商不足...
         */
        int count = submittedCount.incrementAndGet();

        /*
        超过最大的并发任务限制，进行 reject
        依赖的LinkedTransferQueue没有长度限制，因此这里进行控制
        ps:这段逻辑应该是要搬在Queue中处理的...
         */
        if (count > maxSubmittedCount) {
            submittedCount.decrementAndGet();
            getRejectedExecutionHandler().rejectedExecution(command, this);
            //应该加入return语句,否则handler不抛异常时,任务将继续execute,导致submittedCount异常等
            return;
        }

        try {
            super.execute(command);
        } catch (RejectedExecutionException rx) {
            //tomcat
            final EfficientTaskQueue queue = (EfficientTaskQueue) super.getQueue();
            try {
                if (!queue.force(command, timeout, unit)) {
                    submittedCount.decrementAndGet();
                    getRejectedExecutionHandler().rejectedExecution(command, this);
                }
            } catch (InterruptedException x) {
                submittedCount.decrementAndGet();
                getRejectedExecutionHandler().rejectedExecution(command, this);
            }
        }
    }

}
